package org.zjvis.datascience.service.dataprovider;

import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.sql.*;
import java.util.*;
import java.util.stream.Collectors;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Joiner;
import com.google.common.collect.Maps;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.zjvis.datascience.common.exception.DataScienceException;
import org.zjvis.datascience.common.model.*;
import com.google.common.collect.Lists;
import org.zjvis.datascience.common.util.*;
import org.zjvis.datascience.common.util.db.JDBCUtil;

import static org.zjvis.datascience.common.constant.DatabaseConstant.DEFAULT_DATASET_ID;

/**
 * @description GP数据provider 提供数据库连接池
 * @date 2021-11-02
 */
@Service("gpDataProvider")
public class GPDataProvider extends JdbcDataProvider {

    private static final int QUERY_TIME_OUT_SEC = 0;

    private final static Long CACHE_EXPIRE_TIME = 1 * 60 * 60L; // 1 hour

    @Autowired
    private RedisUtil redisUtil;

    public List<String> showDatabases(Long dsId) throws DataScienceException {
        List<String> ret = Lists.newArrayList();
        Connection conn = null;
        try {
            conn = getConn(dsId);
            Statement stat = conn.createStatement();
            stat.setQueryTimeout(SQL_QUERY_TIMEOUT_SECOND);
            ResultSet rs = stat.executeQuery("select datname from pg_database");
            while (rs.next()) {
                ret.add(rs.getString(1));
            }
        } catch (Exception e) {
            logger.error("show databases error, dsId=" + dsId, e);
            throw new DataScienceException("dsId=" + dsId + "\n" + e.getMessage());
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                }
            }
        }
        return ret;
    }

    public List<String> showTables(Long dsId, String db) throws DataScienceException {
        List<String> ret = Lists.newArrayList();
        Connection conn = null;
        try {
            conn = getConn(dsId);
            Statement stat = conn.createStatement();
            stat.setQueryTimeout(SQL_QUERY_TIMEOUT_SECOND);
//          stat.execute("use " + db);
            ResultSet rs = stat.executeQuery("select table_name from information_schema.tables");
            while (rs.next()) {
                ret.add(rs.getString(1));
            }
        } catch (Exception e) {
            logger.error("show tables error, dsId=" + dsId, e);
            throw new DataScienceException("dsId=" + dsId + "\n" + e.getMessage());
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                }
            }
        }
        return ret;
    }

    public List<Column> queryTableMeta(String dataSourceKey, String sql) {
        Connection connection = null;
        Statement stat = null;
        List<Column> columns = Lists.newArrayList();
        try {
            connection = getConn(DEFAULT_DATASET_ID);
            stat = connection.createStatement();
            stat.setQueryTimeout(QUERY_TIME_OUT_SEC);
            ResultSet resultSet = stat.executeQuery(sql);
            for (int i = 1; i <= resultSet.getMetaData().getColumnCount(); i++) {
                Column.ColumnBuilder builder = Column.builder()
                        .index(i)
                        .name(resultSet.getMetaData().getColumnName(i))
                        .label(resultSet.getMetaData().getColumnLabel(i))
                        .type(resultSet.getMetaData().getColumnType(i))
                        .typeName(resultSet.getMetaData().getColumnTypeName(i));
                columns.add(builder.build());
            }
        } catch (SQLException e1) {
            logger.error("query sql {} column meta info failed1, since {}", sql, e1.getMessage());
            throw new DataScienceException("show tables meta error, since {}"+  e1.getMessage());
        } catch (Exception e) {
            throw new DataScienceException("show tables meta error, since {}"+  e.getMessage());
        } finally {
            try {
                stat.close();
                connection.close();
            } catch (SQLException e3) {
                logger.error("close connection failed. since {}", e3.getMessage());
            }
        }
        return columns;
    }


    public JSONArray queryTableMeta(String selectSql){
        Connection conn = null;
        try {
            conn  = getConn(DEFAULT_DATASET_ID);
            Statement st = conn.createStatement();
            ResultSet rs = st.executeQuery(selectSql);
            ResultSetMetaData meta = rs.getMetaData();
            List<String> headerNames = new ArrayList<>();
            JSONArray headers = new JSONArray();
            for (int i = 1; i < meta.getColumnCount() + 1; i++) {
                JSONObject header = new JSONObject();
                String name = meta.getColumnName(i);
                headerNames.add(name);
                if (name.equals("_record_id_")) {
                    continue;
                }
                header.put("name", meta.getColumnName(i));
                header.put("type", meta.getColumnType(i));
                header.put("desc", SqlUtil.changeType(meta.getColumnTypeName(i)));
                headers.add(header);
            }
            return headers;
        }catch (Exception e) {
            logger.error("show tables meta error, since {}" , e.getMessage());
            throw new DataScienceException("show tables meta error, since {}"+  e.getMessage());
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                }
            }
        }
    }

    public Boolean checkTableIfExist(Table table) {
        String sql = "SELECT COUNT(1) FROM " + table.getName();

        Connection conn = null;
        try {
            conn = getConn(table.getDsId());
            Statement stat = conn.createStatement();
            stat.executeQuery(sql);
            return true;
        } catch (Exception e) {
            return false;
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                }
            }
        }
    }

    public String getSql(Table table, AggrConfig config) throws DataScienceException {
        String sql = super.getSql(table, config);
        return sql.replaceAll("`", "\"");
    }

    public void executeSql(String sql) {
        Connection conn = null;
        try {
            conn = getConn(DEFAULT_DATASET_ID);
            Statement st = conn.createStatement();
            executeSql(st, sql);
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            JDBCUtil.close(conn, null, null);
        }
    }

    public <T> T executeQuerySQL(String dataSourceKey, String sql, Class<T> clazz) {
        return executeQuerySQL(dataSourceKey, sql, clazz, false);
    }

    @SuppressWarnings("all")
    @SneakyThrows
    public <T> T executeQuerySQL(String dataSourceKey, String sql, Class<T> clazz, Boolean forceQuery) {
        String md5Code = org.apache.commons.lang.StringUtils.EMPTY;
        if (!forceQuery) {
            //增加缓存层
            MessageDigest md5 = MessageDigest.getInstance("MD5");
            md5.update(StandardCharsets.UTF_8.encode(dataSourceKey + sql + clazz.toString()));
            md5Code = "GP_QUERY_" + String.format("%032x", new BigInteger(1, md5.digest()));
            Object cacheResult = redisUtil.get(md5Code);
            if (ObjectUtil.isNotNull(cacheResult)) {
                //命中后延长过期时间
                redisUtil.expire(md5Code, CACHE_EXPIRE_TIME);
                logger.info(" [executeQuerySQL {}] hit the cache when query sql {} and class {}", dataSourceKey, sql, clazz);
                return (T) cacheResult;
            }
        }
        logger.info(" [executeQuerySQL] got request dataSourceKey {}  and sql {} and return type {}", dataSourceKey, sql, clazz);

        //TODO 检查table的数据源配置
        JSONArray resultArray = this.executeQuerySQL(sql);

        if (ObjectUtil.isNotEmpty(resultArray)) {
            try {
                T result = JsonParseUtil.extractResult(resultArray, clazz);

                if (!forceQuery) redisUtil.set(md5Code, result, CACHE_EXPIRE_TIME);

                return result;

            } catch (Exception e) {
                if (!forceQuery) redisUtil.del(md5Code);
                throw new DataScienceException(String.format(" [executeQuerySQL] failed to execute sql {} and class {}", sql, clazz), e);
            }
        } else {
            if (!forceQuery) redisUtil.del(md5Code);
            return (T) ClazzUtil.initDefaultValue(clazz);
        }
    }

    public JSONArray executeQuerySQL(String sql) {
        //检查SQL
        Connection conn = null;
        Statement stat = null;
        JSONArray queryResult = new JSONArray();
        ResultSet resultSet = null;
        try {
            conn = getConn(DEFAULT_DATASET_ID);
            stat = conn.createStatement();
            stat.setQueryTimeout(QUERY_TIME_OUT_SEC);
            resultSet = stat.executeQuery(sql);
        } catch (SQLException e1) {
            logger.error("query sql {} failed, since {}", sql, e1.getMessage());
            throw new DataScienceException(String.format("query sql {} failed, since {}", sql, e1.getMessage()), e1);
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            ResultSetMetaData metaData = resultSet.getMetaData();
            while (resultSet.next()) {
                LinkedHashMap<String, Object> column = Maps.newLinkedHashMap();
                for (int i = 1; i < metaData.getColumnCount() + 1; i++) {
                    String name = metaData.getColumnName(i);
                    //判断sql select body中是否重名
                    if (column.containsKey(name)) {
                        name = Joiner.on("_").join(name, StringUtil.genLexicalKey(i));
                    }
                    column.put(name, resultSet.getObject(i));
                }
                queryResult.add(column);
            }
        } catch (SQLException e2) {
            logger.error("retrieve sql {} result failed, since {}", sql, e2.getMessage());
            throw new DataScienceException(String.format("retrieve sql {} result failed, since {}", sql, e2.getMessage()), e2);
        } finally {
            JDBCUtil.close(conn, stat, null);
        }
        return queryResult;
    }

    public void executeSql(Statement st, String sql) throws SQLException {
        st.execute(sql);
    }

    public JSONArray executeQuery(String selectSql) {
        JSONArray sqlData = new JSONArray();
        Connection conn = null;
        try {
            conn = getConn(DEFAULT_DATASET_ID);
            Statement st = conn.createStatement();
            sqlData = executeQuery(st, selectSql);
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            JDBCUtil.close(conn, null, null);
        }
        return sqlData;
    }

    public JSONArray executeQuery(Statement st, String selectSql) throws SQLException {
        ResultSet rs = st.executeQuery(selectSql);
        ResultSetMetaData meta = rs.getMetaData();

        JSONArray values = new JSONArray();
        while (rs.next()) {
            JSONObject column = new JSONObject(new LinkedHashMap<>());
            for (int i = 1; i < meta.getColumnCount() + 1; i++) {
                String name = meta.getColumnName(i);
                column.put(name, rs.getObject(name));
            }
            values.add(column);
        }

        return values;
    }

    public Set<String> executeQueryAsOneSet(String selectSql) {
        Set<String> sqlData = new HashSet<>();
        Connection conn = null;
        try {
            conn = getConn(DEFAULT_DATASET_ID);
            Statement st = conn.createStatement();
            sqlData = executeQueryAsOneSet(st, selectSql);
        } catch (Exception e) {
            logger.error(e.getMessage());
        } finally {
            JDBCUtil.close(conn, null, null);
        }
        return sqlData;
    }

    public Set<String> executeQueryAsOneSet(Statement st, String selectSql) throws SQLException {
        Set<String> sqlData = new HashSet<>();
        ResultSet rs = st.executeQuery(selectSql);
        ResultSetMetaData meta = rs.getMetaData();

        while (rs.next()) {
            for (int i = 1; i < meta.getColumnCount() + 1; i++) {
                if (rs.getString(meta.getColumnName(i)) != null) {
                    String str = rs.getString(meta.getColumnName(i));
                    if (str != null && !str.trim().isEmpty()) {
                        sqlData.add(str);
                    }
                }
            }
        }
        return sqlData;
    }

    /**
     * 不需要强制读数据库，拿最新数据
     * @param table
     * @param config
     * @return
     * @throws DataScienceException
     */
    public AggregateResult getData(Table table, AggrConfig config) throws DataScienceException {
        return getData(table, config, false);
    }

    public AggregateResult getData(Table table, AggrConfig config, Boolean forceQuery) throws DataScienceException {
        List<Object[]> ret = Lists.newArrayList();
        AggregateResult aggregateResult = new AggregateResult();
        String sql = StringUtils.EMPTY;
        try {
            sql = getSql(table, config);
        } catch (DataScienceException e1) {
            aggregateResult.setErrorMsg(e1.getApiResult().getMessage());
            return aggregateResult;
        } catch (Exception e2) {
            aggregateResult.setErrorMsg(String.format("something wrong when generate sql, since %s", e2.getMessage()));
            return aggregateResult;
        }
        logger.info("[getData] execute SQL -> {}", sql);
        int dimSize = config.getDimensions().size();
        int measureSize = config.getMeasures().size();

        int index = -1;
        try {
            JSONArray result = this.executeQuery(sql);
            ret = result.toJavaList(Map.class).stream().map(m -> m.values().toArray()).collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("get " + table + " column value error", e);
            throw new DataScienceException(sql + "\n" + e.getMessage(), e);
        }

        List<AggregateColumn> aggrColumns = Lists.newArrayListWithCapacity(dimSize + measureSize);

        for (int i = 0; i < dimSize; i++) {
            AggregateColumn column = new AggregateColumn();
            column.setAggrType(null);
            DimensionConfig dc = config.getDimensions().get(i);
            if (StringUtils.isNotEmpty(dc.getAlias())) {
                column.setName(dc.getAlias());
            } else {
                column.setName(dc.getFieldName());
            }
            column.setIndex(i);
            column.setType(dc.getType());
            aggrColumns.add(column);
        }

        for (int i = 0; i < measureSize; i++) {
            AggregateColumn column = new AggregateColumn();
            MeasureConfig mc = config.getMeasures().get(i);
            column.setAggrType(mc.getAggType());
            column.setName(mc.getAlias());
            column.setIndex(i + dimSize);
            aggrColumns.add(column);
        }

        return aggregateResult.init(aggrColumns, ret.toArray(new Object[ret.size()][]));
    }

}
